package flink_p1

import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 *
 * allowLateness
 *
 *
 * */

object FlinkTest_21_window_watermark_allowLateness {


  /**
   * 使用event time来计算WordCount
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(3)) {
        override def extractTimestamp(element: String): Long = element.split(" ")(0).toLong
      })


    val laterTag = new OutputTag[(String, Int)]("later")

    val resStream: DataStream[(String, Int)] = socketStream.map(data => {
      (data.split(" ")(1), 1)
    }).keyBy(_._1)
      .timeWindow(Time.seconds(5), Time.seconds(5))
      .allowedLateness(Time.seconds(2))
      .sideOutputLateData(laterTag)
      .process(new ProcessWindowFunction[(String, Int), (String, Int), String, TimeWindow] {
        override def process(key: String, context: Context, elements: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
          println(s"win: ${context.window.getStart}   ${context.window.getEnd}")

          var c: Int = 0
          for (elem <- elements) {
            c = c + elem._2
          }
          out.collect(key, c)
        }
      })


    resStream.print("main")

    resStream.getSideOutput(laterTag).print("later")


    env.execute()
  }

}
